package com.bubbles.workspace.sync.service.source.feature;


import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.bubbles.workspace.sync.constant.Constant;
import com.bubbles.workspace.sync.entity.source.metadata.SourceCatalog;
import com.bubbles.workspace.sync.entity.source.metadata.SourceJdbc;
import com.bubbles.workspace.sync.entity.source.metadata.SourceServer;
import com.bubbles.workspace.sync.properties.DorisProperties;
import com.bubbles.workspace.sync.properties.SshProperties;
import com.bubbles.workspace.sync.service.source.constraint.*;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDimensionService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDorisAnalyzeColumnService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceDorisAnalyzeTableService;
import com.bubbles.workspace.sync.service.source.enhance.SourceEnhanceIncrementService;
import com.bubbles.workspace.sync.service.source.metadata.*;
import com.bubbles.workspace.sync.util.JdbcUtil;
import com.bubbles.workspace.sync.util.SshManager;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.annotation.Lazy;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Date;

/**
 * <p>description: 源数据源-测试服务 </p>
 *
 * @author CryptoNeedle
 * @date 2025-06-23
 */
@DS("pg")
@Slf4j
@Service
public class SourceTestService {
    
    @Resource
    @Lazy
    private SourceTestService sourceTestService;
    
    @Resource
    private SourceJdbcService sourceJdbcService;
    
    @Resource
    private SourceServerService sourceServerService;
    
    @Resource
    private SourceCatalogService sourceCatalogService;
    
    @Resource
    private SourceDatabaseService sourceDatabaseService;
    
    @Resource
    private SourceTableService sourceTableService;
    
    @Resource
    private SourceColumnService sourceColumnService;
    
    @Resource
    private SourceConstraintPrimaryService sourceConstraintPrimaryService;
    
    @Resource
    private SourceConstraintUniqueService sourceConstraintUniqueService;
    
    @Resource
    private SourceConstraintForeignService sourceConstraintForeignService;
    
    @Resource
    private SourceConstraintCheckService sourceConstraintCheckService;
    
    @Resource
    private SourceIndexUniqueService sourceIndexUniqueService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeTableService sourceEnhanceDorisAnalyzeTableService;
    
    @Resource
    private SourceEnhanceDorisAnalyzeColumnService sourceEnhanceDorisAnalyzeColumnService;
    
    @Resource
    private SourceEnhanceIncrementService sourceEnhanceIncrementService;
    
    @Resource
    private SourceEnhanceDimensionService sourceEnhanceDimensionService;
    
    @Resource
    private SshProperties sshProperties;
    
    @Resource
    private DorisProperties dorisProperties;
    
    @Resource
    private SshManager sshManager;
    
    @Resource
    private JdbcTemplate jdbcTemplate;
    
    @Async
    @DSTransactional
    public void testCatalogConnectionAsync(SourceCatalog sourceCatalog) {
        sourceTestService.testCatalogConnection(sourceCatalog);
    }
    
    @DSTransactional
    public boolean testCatalogConnection(SourceCatalog sourceCatalog) {
        boolean serverConnected;
        boolean jdbcConnected;
        boolean dorisConnected;
        
        String sourceCatalogName = sourceCatalog.getSourceCatalogName();
        SourceJdbc sourceJdbc = sourceCatalog.getSourceJdbc();
        SourceServer sourceServer = sourceJdbc.getSourceServer();
        
        log.info("[TEST] {} =======================> [开始]", sourceCatalogName);
        
        serverConnected = sourceTestService.testServerConnection(sourceServer);
        log.info("[TEST] {} -> 连接服务器 => {}", sourceCatalogName, serverConnected);
        
        jdbcConnected = sourceTestService.testJdbcConnection(sourceJdbc);
        log.info("[TEST] {} -> 连接JDBC  => {}", sourceCatalogName, jdbcConnected);
        
        dorisConnected = sourceTestService.testDorisConnection(sourceCatalog);
        log.info("[TEST] {} -> 连接Doris => {}", sourceCatalogName, dorisConnected);
        
        log.info("[TEST] {} =======================> [结束]", sourceCatalogName);
        
        // 目前不强制要求 Doris Catalog 可连通
        return jdbcConnected;
    }
    
    @DSTransactional
    public boolean testServerConnection(SourceServer sourceServer) {
        boolean serverConnected = false;
        
        // 测试
        if (SshManager.getSuccessForwardMap().containsKey(sourceServer.getInetAddress())) {
            serverConnected = true;
        } else {
            String host = sourceServer.getHost();
            Integer port = sourceServer.getPort();
            
            if (sshProperties.getEnabled()) {
                try {
                    // 重新代理
                    serverConnected = sshManager.forward(host, port);
                } catch (Exception e) {
                    log.error("[TEST] 连接服务器 -> 异常 => {}", e.getMessage());
                }
            } else {
                try (Socket socket = new Socket()) {
                    socket.connect(new InetSocketAddress(host, port), sshProperties.getTimeout());
                    serverConnected = true;
                } catch (IOException e) {
                    log.error("[TEST] 连接服务器 -> 异常 => {}", e.getMessage());
                }
            }
        }
        
        // 保存
        sourceServer.setConnected(serverConnected);
        if (serverConnected) {
            sourceServer.setConnectedDatetime(new Date());
        }
        sourceServerService.save(sourceServer);
        
        return serverConnected;
        
    }
    
    @DS("pg")
    @DSTransactional
    public boolean testJdbcConnection(SourceJdbc sourceJdbc) {
        String version = null;
        
        // 测试
        try {
            JdbcTemplate jdbcTemplate = JdbcUtil.getJdbcTemplate(sourceJdbc);
            version = jdbcTemplate.queryForObject(sourceJdbc.getDatabaseType()
                                                            .getDatabase()
                                                            .getDatabaseVersionSql(), String.class);
        } catch (Exception e) {
            log.error("[TEST] 连接JDBC -> 异常 => {}", e.getMessage());
        }
        
        // 保存
        boolean jdbcConnected = StringUtils.isNotEmpty(version);
        sourceJdbc.setConnected(jdbcConnected);
        if (jdbcConnected) {
            sourceJdbc.setDatabaseVersion(version);
            sourceJdbc.setConnectedDatetime(new Date());
        }
        sourceJdbcService.save(sourceJdbc);
        
        return jdbcConnected;
    }
    
    @DS("doris")
    @DSTransactional
    public boolean testDorisConnection(SourceCatalog sourceCatalog) {
        boolean dorisConnected = false;
        
        // 测试
        String transformSystemCode = sourceCatalog.getTransform().getTransformSystemCode();
        if (!StringUtils.isEmpty(transformSystemCode)) {
            try {
                String dropTestCatalogSql = String.format("DROP CATALOG IF EXISTS %s%s", Constant.TEST_CATALOG_PREFIX, transformSystemCode);
                jdbcTemplate.execute(dropTestCatalogSql);
                
                String createTestCatalogSql = sourceCatalogService.createCatalogSql(sourceCatalog, true);
                jdbcTemplate.execute(createTestCatalogSql);
                
                dropTestCatalogSql = String.format("DROP CATALOG %s%s", Constant.TEST_CATALOG_PREFIX, transformSystemCode);
                jdbcTemplate.execute(dropTestCatalogSql);
                
                dorisConnected = true;
            } catch (Exception e) {
                log.error("[TEST] 连接Doris -> 异常 => {}", e.getMessage());
            }
        }
        
        // 保存
        SourceCatalog.Doris sourceCatalogDoris = sourceCatalog.getDoris();
        sourceCatalogDoris.setDorisConnected(dorisConnected);
        if (dorisConnected) {
            sourceCatalogDoris.setDorisConnectedDatetime(new Date());
        }
        sourceCatalogService.save(sourceCatalog);
        
        return dorisConnected;
    }
}